package com.xujian.flink.sink.file;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;

/**
 * @author star xu
 * @date 2023/1/13 15:15
 * @Description:
 * @Slogan: 致敬大师, 致敬未来的自己
 */
public class FileSinkMain {
    public static void main(String[] args) throws Exception {
        // 1. 创建流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 读取文本
        FileSource<String> source = FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path("input/words.txt")).build();
        DataStreamSource<String> lineDSS = env.fromSource(source, WatermarkStrategy.noWatermarks(), "file source");
        //3 sink to file
        OutputFileConfig outputFileConfig = OutputFileConfig
                .builder()
                .withPartPrefix("prefix")
                .withPartSuffix(".txt")
                .build();
        final FileSink<String> fileSink = FileSink
                .forRowFormat(new Path("D:/ft"), new SimpleStringEncoder<String>("UTF-8"))
                .withOutputFileConfig(outputFileConfig)
                .build();
        lineDSS.sinkTo(fileSink);
        // 4. 执行
        env.execute();
    }
}
